﻿
Imports System.Threading
Imports std

Public Class threadpool
    Public Const defaultStopWaitSeconds As Int64 = 30
    Public maxIdleRound As Int64 = 8
    Public idletimeMS As Int64 = 1

    'use pointer of object to avoid copy the object again
    Private queue As queue(Of pair(Of ParameterizedThreadStart, pointer(Of Object))) = Nothing
    Private threads As vector(Of Thread) = Nothing
    Private work As ParameterizedThreadStart = Nothing
    Private workingThread As Int64 = 0
    Private stopsignal As Boolean = False

    Private Function queueLengthImpl() As Int64
        Return queue.size()
    End Function

    Public Function queueLength() As Int64
        Dim rtn As Int64
        SyncLock queue
            rtn = queueLengthImpl()
        End SyncLock
        Return rtn
    End Function

    Public Function idle() As Boolean
        Dim rtn As Boolean
        SyncLock queue
            rtn = (Interlocked.Read(workingThread) = 0 AndAlso queueLengthImpl() = 0)
        End SyncLock

        Return rtn
    End Function

    Public Sub insertJob(ByVal obj As Object)
        insertJob(work, obj)
    End Sub

    Public Overridable Sub insertJob(ByVal work As ParameterizedThreadStart, ByVal obj As Object)
        SyncLock queue
            If Not stopsignal Then
                queue.push(make_pair(work, New pointer(Of Object)(obj)))
            Else
                raiseError("should not insertJob after stop.")
            End If
        End SyncLock
    End Sub

#If RETIRED Then
    Private Sub idle(ByRef idleround As Int64)
        'sleep idletime * idleround if no job arrived.
        'idleround <= maxIdleRound
        idleround += 1
        If idleround > maxIdleRound Then
            idleround = maxIdleRound
        End If
        sleepTicks(idletimeticks * idleround)
    End Sub
#End If
    Private Sub idle(ByVal __NOT_IN_USE__ As Int64)
        sleep(idletimeMS)
    End Sub

    Public Sub workthread()
        Dim idleround As Int64 = 0
        While True
            Dim obj As pair(Of ParameterizedThreadStart, pointer(Of Object)) = Nothing
            Dim suc As Boolean = False

            If Not queue.empty() Then
                SyncLock queue
                    If Not queue.empty() Then
                        suc = True
                        obj = queue.front()
                        Interlocked.Increment(workingThread)
                        queue.pop()
                    End If
                End SyncLock
            End If
            If suc Then
                If Not obj.first Is Nothing Then
                    Try
                        obj.first(+(obj.second))
                    Catch ex As ThreadAbortException
                        raiseError("thread abort.", errorHandle.errorType.warning)
                    Catch ex As Exception
                        raiseError("caught unhandled exception in thread call, ex " + ex.Message + _
                                   ", call stack " + ex.StackTrace, errorHandle.errorType.critical)
                    End Try
                Else
                    raiseError("do not get workthreadDelegate from job queue, ignore this request.", _
                               errorHandle.errorType.warning)
                End If
                Interlocked.Decrement(workingThread)
                idleround = 0
            Else
                idle(idleround)
            End If
        End While
    End Sub

    Public Property threadCount() As Int64
        Get
            Return threads.size()
        End Get
        Set(ByVal value As Int64)
            While threads.size() < value
                Dim th As Thread = Nothing
                th = New Thread(AddressOf workthread)
                th.Name() = "THREADPOOL_WORKTHREAD"
                th.Start()
                threads.push_back(th)
            End While
        End Set
    End Property

    Private Sub initial()
        queue = New queue(Of pair(Of ParameterizedThreadStart, pointer(Of Object)))()
        threads = New vector(Of Thread)()
    End Sub

    Public Sub New(ByVal workthread As ParameterizedThreadStart, Optional ByVal threadCount As Int64 = 0)
        assert(Not workthread Is Nothing, "cannot work without a ParameterizedThreadStart.")
        work = workthread
        initial()
        start(threadCount)
    End Sub

    Public Sub New(ByVal threadCount As Int64)
        initial()
        start(threadCount)
    End Sub

    Public Sub New()
        initial()
        start()
    End Sub

    Public Sub start(ByVal threadCount As Int64)
        If Me.threadCount() > threadCount Then
            [stop]()
        End If
        Me.threadCount() = threadCount
        stopsignal = False
    End Sub

    Public Sub start()
#If PocketPC OrElse Smartphone Then
        start(ifoper(threadCount() <= 0, 2, threadCount()))
#Else
        start(ifoper(threadCount() <= 0 _
                     , ifoper(Environment.ProcessorCount() < 2, 2 _
                              , Convert.ToInt64(Environment.ProcessorCount())) _
                     , threadCount()))
#End If
    End Sub

    Public Sub [stop](Optional ByVal stopWaitSeconds As Int64 = defaultStopWaitSeconds)
        stopsignal = True
        If stopWaitSeconds <= 0 Then
            stopWaitSeconds = defaultStopWaitSeconds
        End If
        If Not threads Is Nothing AndAlso threads.size() > 0 Then
            raiseError("threadpool starts to stop, waiting for all jobs in queue.")
            While Not idle() AndAlso stopWaitSeconds > 0
                sleepSeconds(1)
                stopWaitSeconds -= 1
            End While
            raiseError("threadpool finished stopping.")
            Dim i As Int64
            For i = 0 To threadCount() - 1
#If Not (PocketPC OrElse Smartphone) Then
                If Not threads(i) Is Nothing AndAlso threads(i).IsAlive() Then
                    threads(i).Abort()
                    threads(i).Join()
                End If
#Else
                If Not threads(i) Is Nothing Then
                    Try
                        threads(i).Abort()
                        threads(i).Join()
                    Catch
                    End Try
                End If
#End If
            Next
            threads.clear()
        End If
    End Sub
End Class

Public Class adaptingThreadPool
    Inherits threadpool

    Public adaptingLimit As Int64 = 32
    Public Const threadUpperLimit As Int64 = 512

    Public Overrides Sub insertJob(ByVal work As ParameterizedThreadStart, ByVal obj As Object)
        MyBase.insertJob(work, obj)
        If queueLength() > adaptingLimit AndAlso threadCount() < threadUpperLimit Then
            threadCount() >>= 1
        End If
    End Sub
End Class